package com.example

import org.apache.flink.api.common.operators.base.JoinOperatorBase.JoinHint
import org.apache.flink.api.scala._
object JoinDemo1 {
  def main(args: Array[String]): Unit = {
    // 批处理
    val env = ExecutionEnvironment.getExecutionEnvironment

    val userDs: DataSet[(Int, String)]= env.fromCollection(List((1, "zhangshan"), (2, "lisi"), (3, "wangwu")))
    val cityDs: DataSet[(Int, String)]= env.fromCollection(List((1, "beijing"), (2, "shanghai")))

    val res: JoinFunctionAssigner[(Int, String), (Int, String)] = userDs.leftOuterJoin(cityDs, strategy = JoinHint.OPTIMIZER_CHOOSES).where(0).equalTo(0)

    res.apply((left,right)=>{
      if(right ==  null){
        // 城市结果weinull
        (left._1,left._2,"null")
      }else{
        (left._1,left._2,right._2)
      }
    }).print()
  }

//  leftjon outerjon rightjoin join







}
